﻿using Confluent.Kafka;
using MySqlSync.Core.Interfaces;
using MySqlSync.Core.Models;
using System.Text.Json;

namespace MySqlSync.Kafka
{
    /// <summary>
    /// Kafka消费者服务
    /// </summary>
    public class KafkaConsumerService : IKafkaConsumerService
    {
        private readonly IConsumer<string, string> _consumer;
        private readonly string _topic;
        private readonly JsonSerializerOptions _jsonOptions = new JsonSerializerOptions
        {
            PropertyNameCaseInsensitive = true
        };
        private bool _isDisposed = false;

        /// <summary>
        /// 构造函数
        /// </summary>
        public KafkaConsumerService(KafkaConfig config)
        {
            if (config == null)
                throw new ArgumentNullException(nameof(config));

            if (string.IsNullOrEmpty(config.BootstrapServers))
                throw new ArgumentException("Kafka bootstrap servers 不能为空", nameof(config));

            if (string.IsNullOrEmpty(config.Topic))
                throw new ArgumentException("Kafka topic 不能为空", nameof(config));

            if (string.IsNullOrEmpty(config.GroupId))
                throw new ArgumentException("Kafka group id 不能为空", nameof(config));

            _topic = config.Topic;

            var consumerConfig = new ConsumerConfig
            {
                BootstrapServers = config.BootstrapServers,
                GroupId = config.GroupId,
                AutoOffsetReset = AutoOffsetReset.Earliest,
                EnableAutoCommit = false,
                FetchMinBytes = 1024,
                FetchWaitMaxMs = 500,
                //MaxPollRecords = 100
            };

            _consumer = new ConsumerBuilder<string, string>(consumerConfig).Build();
            _consumer.Subscribe(_topic);
        }

        /// <summary>
        /// 消费消息
        /// </summary>
        public async Task ConsumeAsync(Func<SyncRecord, Task> handleMessage, CancellationToken cancellationToken)
        {
            if (handleMessage == null)
                throw new ArgumentNullException(nameof(handleMessage));

            if (_isDisposed)
                throw new ObjectDisposedException(nameof(KafkaConsumerService));

            while (!cancellationToken.IsCancellationRequested)
            {
                try
                {
                    var consumeResult = _consumer.Consume(cancellationToken);

                    if (consumeResult.IsPartitionEOF)
                    {
                        Console.WriteLine($"已到达分区末尾: {consumeResult.TopicPartitionOffset}");
                        continue;
                    }

                    if (consumeResult.Message == null)
                        continue;

                    var record = JsonSerializer.Deserialize<SyncRecord>(consumeResult.Message.Value, _jsonOptions);

                    if (record != null)
                    {
                        await handleMessage(record);
                        // 手动提交偏移量，确保消息处理成功后再提交
                        _consumer.Commit(consumeResult);
                    }
                    else
                    {
                        Console.WriteLine($"无法解析消息: {consumeResult.Message.Value}");
                        // 可以选择提交偏移量跳过无效消息
                        _consumer.Commit(consumeResult);
                    }
                }
                catch (OperationCanceledException)
                {
                    // 预期的取消操作，不处理
                    break;
                }
                catch (ConsumeException ex)
                {
                    Console.WriteLine($"消费错误: {ex.Error.Reason}");
                    // 非致命错误，等待后重试
                    await Task.Delay(1000, cancellationToken);
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"处理消息时发生错误: {ex.Message}");
                    // 发生错误时等待后重试
                    await Task.Delay(1000, cancellationToken);
                }
            }
        }


        /// <summary>
        /// 释放资源
        /// </summary>
        public async ValueTask DisposeAsync()
        {
            if (!_isDisposed)
            {
                _consumer?.Unsubscribe();
                _consumer?.Close();
                _consumer?.Dispose();
                _isDisposed = true;
            }

            await Task.CompletedTask;
        }
    }

}
